package cn.kgc.gmall.app.func;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * 自定义反序列化器
 */
public class MyDeserializationSchemaFunction implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        // 强转为struct类型
        Struct value = (Struct) sourceRecord.value();
        // 获取表结构体
        Struct source = value.getStruct("source");
        // 获取数据库
        String db = source.getString("db");
        // 获取表
        String table = source.getString("table");
        // 获取数据的结构体
        Struct after = value.getStruct("after");
        // String op = value.getString("op");
        // 获取操作类型
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if (type.equals("create")){
            type = "insert";
        }
        // 获取操作时间
        Long ts_ms = value.getInt64("ts_ms");
        // 转化为json对象
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("database",db);
        jsonObject.put("table",table);
        jsonObject.put("type",type);
        jsonObject.put("ts",ts_ms);

        JSONObject dateJson = new JSONObject();
        // 动态获取数据
        // 既要获取名字还要获取值
        if (after != null){
            Schema schema = after.schema();
            // 遍历出结构体中的所有的字符按
            for (Field field : schema.fields()) {
                String name = field.name();
                String val = "";
                if (after.get(field) != null){
                     val =  after.get(field).toString();

                }
                dateJson.put(name,val);
            }
        }
        // 将数据中的json对象装入到父级json中
        jsonObject.put("data",dateJson);

        collector.collect(jsonObject.toJSONString());
    }


    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}
